package project2;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.jobcontrol.ControlledJob;
import org.apache.hadoop.mapreduce.lib.jobcontrol.JobControl;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class ExcelMR extends Configured implements Tool {
	@Override
	public int run(String[] args) throws Exception {
		if (args.length != 3) {
			System.out.println("usage ...");
			return -1;
		}
		Configuration config = getConf();
		FileSystem fs = FileSystem.get(config);
		Path path = new Path(args[2]);
		if (fs.exists(path)) {
			fs.delete(path, true);
		}
		Path path1 = new Path(args[2] + "_");;
		if (fs.exists(path1)) {
			fs.delete(path1, true);
		}
		
		Job job1 = Job.getInstance(config, "Excel->Text");
		job1.setJarByClass(getClass());
		// 添加到内存中的缓存文件，可添加多个
		job1.addCacheFile(new URI(args[1]));
		//
		job1.setMapperClass(ExcelMapper.class);
		job1.setMapOutputKeyClass(Text.class);
		job1.setMapOutputValueClass(LongWritable.class);
		//
		job1.setReducerClass(ExcelReducer.class);
		job1.setOutputKeyClass(Text.class);
		job1.setOutputValueClass(LongWritable.class);
		//
		job1.setInputFormatClass(ExcelInputFormat.class);
		ExcelInputFormat.addInputPath(job1, new Path(args[0]));
		FileOutputFormat.setOutputPath(job1, path);
		
		
		
		Job job2 = Job.getInstance(config, "通话记录统计");
		job2.setJarByClass(getClass());
		//
		job2.setMapperClass(ExcelCountMapper2.class);
		job2.setMapOutputKeyClass(PhoneExcelWritable.class);
		job2.setMapOutputValueClass(NullWritable.class);
		//
		job2.setNumReduceTasks(2);
		job2.setPartitionerClass(ExcelPartitioner.class);
		job2.setGroupingComparatorClass(FlagGroupingComparator.class);
		//job2.setSortComparatorClass(DurationSortComparator.class);
		//
		job2.setReducerClass(ExcelCountReducer2.class);
		job2.setOutputKeyClass(PhoneExcelWritable.class);
		job2.setOutputValueClass(NullWritable.class);
		FileInputFormat.addInputPath(job2, path);// 前一个job的输入，是后一个job的输出！
		FileOutputFormat.setOutputPath(job2, path1);// out001_1
		
		ControlledJob controlledJob1 = new ControlledJob(config);
		controlledJob1.setJob(job1);		
		ControlledJob controlledJob2 = new ControlledJob(config);
		controlledJob2.setJob(job2);
		controlledJob2.addDependingJob(controlledJob1);
		
		JobControl jobControl = new JobControl("Excel");
		jobControl.addJob(controlledJob1);
		jobControl.addJob(controlledJob2);
		
		// 启动jobControl
		new Thread(jobControl).start();
		//判断是否执行成功
		int code = 0;
		while (true) {
			if (jobControl.allFinished()) {// 判断是否是都执行完成
				List<ControlledJob> list1 = jobControl.getSuccessfulJobList();
				System.out.println("成功的个数：" + list1.size());
				break;
			}
			List<ControlledJob> list2 = jobControl.getFailedJobList();
			if (list2.size() > 0) {// 有失败的
				System.out.println("失败的个数：" + list2.size());
				code = -1;
				break;
			}
		}
		return code;
	}

	public static void main(String[] args) throws Exception {
		int code = ToolRunner.run(new ExcelMR(), args);
		System.exit(code);
	}

	//1:Mapper:
	public static class ExcelMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
		private Map<String, String> contacts = new HashMap<>();

		@Override
		protected void setup(Mapper<LongWritable, Text, Text, LongWritable>.Context context)
				throws IOException, InterruptedException {
			// 获取缓存文件，并将值放入Map<>中
			URI[] cache = context.getCacheFiles();
			if (cache.length > 0) {
				String path = cache[0].getPath(); // 只读取了一个缓存文件
				String fileName = path.substring(path.lastIndexOf("/") + 1);
				File file = new File(fileName);
				if (file.exists()) {
					System.out.println("添加的缓存文件是：" + fileName);
					BufferedReader br = new BufferedReader(new FileReader(file));
					String line = "";
					while ((line = br.readLine()) != null) {
						String[] str = line.split("\t");
						contacts.put(str[1], str[0]);
					}
					br.close();
				}
			}
		}

		@Override
		public void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context)
				throws IOException, InterruptedException {
			// 主叫：1 被叫：0
			String[] string = value.toString().split("\t");
			String phone = string[1];
			if (phone.startsWith("86")) {
				phone = phone.substring(2);
			}
			if (contacts.containsKey(phone)) {
				String name = contacts.get(phone);
				int flag = 0;
				if (string[0].endsWith("主叫")) {
					flag = 1;
				}
				String month = string[2].split("-")[1];
				int duration = Integer.parseInt(string[3]);
				context.write(new Text(month + "\t" + flag + "\t" + name + "\t" + phone), new LongWritable(duration));
			}
		}
	}
	
	//2:Reducer
	public static class ExcelReducer extends Reducer<Text, LongWritable,Text, LongWritable> {
		private LongWritable value4 = new LongWritable();
		@Override
		protected void reduce(Text key3, Iterable<LongWritable> value3,
				Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException, InterruptedException {
			long ii = 0;
			for (LongWritable longWritable : value3) {
				ii += longWritable.get();
			}
			value4.set(ii);
			System.out.println("-----66666665");
			context.write(key3, value4);
		}
	}
	
	
	
	//:Mapper2
	public static class ExcelCountMapper2 extends Mapper<Text, LongWritable, PhoneExcelWritable, NullWritable>{
		 private PhoneExcelWritable phoneExcel;
		@Override
		protected void map(Text key, LongWritable value,
				Mapper<Text, LongWritable, PhoneExcelWritable, NullWritable>.Context context)
				throws IOException, InterruptedException {
			String[] strings = key.toString().split("\t");
			String month = strings[0];
			String flag = strings[1];
			String name = strings[2];
			String phone = strings[3];

			System.out.println("------------------00000");
			int duration = (int) value.get();

			System.out.println("------------------100000"+duration);
			System.out.println("------------------1");
			phoneExcel = new PhoneExcelWritable(month, flag, name, phone, duration);

			System.out.println("------------------2");
			context.write(phoneExcel, NullWritable.get());

			System.out.println("------------------3");
		}
	}

	//2:分区
	public static class ExcelPartitioner extends Partitioner<PhoneExcelWritable, NullWritable>{
		@Override
		public int getPartition(PhoneExcelWritable key, NullWritable value, int numPartition) {
			String month = key.getMonth();
			int pp= Integer.parseInt(month)%numPartition;
			return pp;
		}			
	}
	
	//4:
	public static class FlagGroupingComparator extends WritableComparator{		
		public FlagGroupingComparator() {//必须创建构造方法
			super(PhoneExcelWritable.class,true); //后边的参数代表：是否进行实例化（不能省！！！
		}
		@Override
		public int compare(WritableComparable a, WritableComparable b) {
			PhoneExcelWritable aa = (PhoneExcelWritable) a;
			PhoneExcelWritable bb = (PhoneExcelWritable) b;
			return aa.getFlag().compareTo(bb.getFlag());		
		}
	}

//	//5:
//	public static class DurationSortComparator extends WritableComparator{		
//		public DurationSortComparator() {
//			super(LongWritable.class,true);			
//		}
//
//		@Override
//		public int compare(WritableComparable a, WritableComparable b) {
//			LongWritable aa = (LongWritable) a;
//			LongWritable bb = (LongWritable) b;
//			System.out.println("aa:"+aa.get()+",bb:"+bb.get());
//			return (int) (bb.get()-aa.get());
//		}
//	}
	
	public static class ExcelCountReducer2 extends Reducer<PhoneExcelWritable, NullWritable, PhoneExcelWritable, NullWritable> {
//		Map<String, Long> map = new HashMap<>();
		int aaaa=0;
		
		@Override
			protected void reduce(PhoneExcelWritable key3, Iterable<NullWritable> value3,
					Reducer<PhoneExcelWritable, NullWritable, PhoneExcelWritable, NullWritable>.Context context)
					throws IOException, InterruptedException {
			aaaa+=1;
			System.out.println("!!!!!!!!!!!!!  aaaa:"+aaaa);
			for (NullWritable nullWritable : value3) {
				context.write(key3, NullWritable.get());
			}
			}

//		@Override
//		protected void cleanup(Reducer<Text, LongWritable, Text, Text>.Context context)
//				throws IOException, InterruptedException {
//			for (Entry<String, Long> entry : map.entrySet()) {
//				long duration = entry.getValue();
//				String minutes = duration / 60 + "分" + duration % 60 + "秒";
//				context.write(new Text(entry.getKey()), new Text(minutes));// 03 147分9秒
//			}
//		}

	}

}
